package com.atguigu.gmall.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.app.BaseAppV1;
import com.atguigu.gmall.realtime.bean.TradeSkuOrderBean;
import com.atguigu.gmall.realtime.common.Constant;
import com.atguigu.gmall.realtime.function.DimAsyncFunction;
import com.atguigu.gmall.realtime.util.AtguiguUtil;
import com.atguigu.gmall.realtime.util.FlinkSinkUtil;
import org.apache.commons.beanutils.BeanUtils;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.math.BigDecimal;
import java.time.Duration;
import java.util.concurrent.TimeUnit;

/**
 * @Author lzc
 * @Date 2022/12/13 08:30
 */
public class Dws_09_DwsTradeSkuOrderWindow_Async extends BaseAppV1 {
    public static void main(String[] args) {
        new Dws_09_DwsTradeSkuOrderWindow_Async().init(
            4009,
            2,
            "Dws_09_DwsTradeSkuOrderWindow",
            Constant.TOPIC_DWD_TRADE_ORDER_DETAIL
        );
    }
    
    @Override
    protected void handle(StreamExecutionEnvironment env,
                          DataStreamSource<String> stream) {
        //        2. 解析成 pojo 类型
        SingleOutputStreamOperator<TradeSkuOrderBean> beanStream = parseToPojo(stream);
        //        3. 按照详情 id, 去重
        beanStream = distinctByOrderDetailId(beanStream);
        //        4. 分组,开窗, 聚和
        //        key: sku_id
        SingleOutputStreamOperator<TradeSkuOrderBean> beanStreamWithoutDims = windowAndAgg(beanStream);
        //        5. 补充维度信息
        //        sku spu  tm c3 c2 c1
        SingleOutputStreamOperator<TradeSkuOrderBean> beanStreamWithDims = joinDims(beanStreamWithoutDims);
        
        //
        //        6. 写处到 clickhouse 中
        writeToClickHouse(beanStreamWithDims);
    }
    
    private void writeToClickHouse(SingleOutputStreamOperator<TradeSkuOrderBean> beanStreamWithDims) {
        beanStreamWithDims.addSink(FlinkSinkUtil.getClickHouseSink("dws_trade_sku_order_window", TradeSkuOrderBean.class));
    }
    
    private SingleOutputStreamOperator<TradeSkuOrderBean> joinDims(
        SingleOutputStreamOperator<TradeSkuOrderBean> stream) {
        SingleOutputStreamOperator<TradeSkuOrderBean> skuInfoStream = AsyncDataStream
            .unorderedWait(
                stream, // 需要做异步处理的流
                new DimAsyncFunction<TradeSkuOrderBean>() {
                    
                    @Override
                    public String getTable() {
                        return "dim_sku_info";
                    }
                    
                    @Override
                    public String getId(TradeSkuOrderBean input) {
                        return input.getSkuId();
                    }
                    
                    @Override
                    public void addDim(TradeSkuOrderBean input,
                                       JSONObject dim) {
                        
                        input.setSkuName(dim.getString("SKU_NAME"));
                        
                        input.setSpuId(dim.getString("SPU_ID"));
                        input.setTrademarkId(dim.getString("TM_ID"));
                        input.setCategory3Id(dim.getString("CATEGORY3_ID"));
                        
                    }
                },
                60,
                TimeUnit.SECONDS
            );
        
        SingleOutputStreamOperator<TradeSkuOrderBean> spuInfoStream = AsyncDataStream
            .unorderedWait(
                skuInfoStream, // 需要做异步处理的流
                new DimAsyncFunction<TradeSkuOrderBean>() {
                    
                    @Override
                    public String getTable() {
                        return "dim_spu_info";
                    }
                    
                    @Override
                    public String getId(TradeSkuOrderBean input) {
                        return input.getSpuId();
                    }
                    
                    @Override
                    public void addDim(TradeSkuOrderBean input,
                                       JSONObject dim) {
                        
                        input.setSpuName(dim.getString("SPU_NAME"));
                    }
                },
                60,
                TimeUnit.SECONDS
            );
        
        SingleOutputStreamOperator<TradeSkuOrderBean> tmStream = AsyncDataStream
            .unorderedWait(
                spuInfoStream, // 需要做异步处理的流
                new DimAsyncFunction<TradeSkuOrderBean>() {
                    
                    @Override
                    public String getTable() {
                        return "dim_base_trademark";
                    }
                    
                    @Override
                    public String getId(TradeSkuOrderBean input) {
                        return input.getTrademarkId();
                    }
                    
                    @Override
                    public void addDim(TradeSkuOrderBean input,
                                       JSONObject dim) {
                        
                        input.setTrademarkName(dim.getString("TM_NAME"));
                    }
                },
                60,
                TimeUnit.SECONDS
            );
        
        
        SingleOutputStreamOperator<TradeSkuOrderBean> c3Stream = AsyncDataStream
            .unorderedWait(
                tmStream, // 需要做异步处理的流
                new DimAsyncFunction<TradeSkuOrderBean>() {
                    
                    @Override
                    public String getTable() {
                        return "dim_base_category3";
                    }
                    
                    @Override
                    public String getId(TradeSkuOrderBean input) {
                        return input.getCategory3Id();
                    }
                    
                    @Override
                    public void addDim(TradeSkuOrderBean input,
                                       JSONObject dim) {
                        
                        input.setCategory3Name(dim.getString("NAME"));
                        input.setCategory2Id(dim.getString("CATEGORY2_ID"));
                    }
                },
                60,
                TimeUnit.SECONDS
            );
        
        SingleOutputStreamOperator<TradeSkuOrderBean> c2Stream = AsyncDataStream
            .unorderedWait(
                c3Stream, // 需要做异步处理的流
                new DimAsyncFunction<TradeSkuOrderBean>() {
                    
                    @Override
                    public String getTable() {
                        return "dim_base_category2";
                    }
                    
                    @Override
                    public String getId(TradeSkuOrderBean input) {
                        return input.getCategory2Id();
                    }
                    
                    @Override
                    public void addDim(TradeSkuOrderBean input,
                                       JSONObject dim) {
                        
                        input.setCategory2Name(dim.getString("NAME"));
                        input.setCategory1Id(dim.getString("CATEGORY1_ID"));
                    }
                },
                60,
                TimeUnit.SECONDS
            );
        
        return AsyncDataStream
            .unorderedWait(
                c2Stream, // 需要做异步处理的流
                new DimAsyncFunction<TradeSkuOrderBean>() {
                    
                    @Override
                    public String getTable() {
                        return "dim_base_category1";
                    }
                    
                    @Override
                    public String getId(TradeSkuOrderBean input) {
                        return input.getCategory1Id();
                    }
                    
                    @Override
                    public void addDim(TradeSkuOrderBean input,
                                       JSONObject dim) {
                        
                        input.setCategory1Name(dim.getString("NAME"));
                    }
                },
                60,
                TimeUnit.SECONDS
            );
        
        
    }
    
    private SingleOutputStreamOperator<TradeSkuOrderBean> windowAndAgg(
        SingleOutputStreamOperator<TradeSkuOrderBean> beanStream) {
        return beanStream
            .assignTimestampsAndWatermarks(
                WatermarkStrategy
                    .<TradeSkuOrderBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                    .withTimestampAssigner((bean, ts) -> bean.getTs())
            )
            .keyBy(TradeSkuOrderBean::getSkuId)
            .window(TumblingEventTimeWindows.of(Time.seconds(5)))
            .reduce(
                new ReduceFunction<TradeSkuOrderBean>() {
                    @Override
                    public TradeSkuOrderBean reduce(TradeSkuOrderBean value1,
                                                    TradeSkuOrderBean value2) throws Exception {
                        value1.setOriginalAmount(value1.getOriginalAmount().add(value2.getOriginalAmount()));
                        value1.setOrderAmount(value1.getOrderAmount().add(value2.getOrderAmount()));
                        value1.setActivityAmount(value1.getActivityAmount().add(value2.getActivityAmount()));
                        value1.setCouponAmount(value1.getCouponAmount().add(value2.getCouponAmount()));
                        return value1;
                    }
                },
                new ProcessWindowFunction<TradeSkuOrderBean, TradeSkuOrderBean, String, TimeWindow>() {
                    @Override
                    public void process(String key,
                                        Context ctx,
                                        Iterable<TradeSkuOrderBean> elements,
                                        Collector<TradeSkuOrderBean> out) throws Exception {
                        TradeSkuOrderBean bean = elements.iterator().next();
                        bean.setStt(AtguiguUtil.toDatTime(ctx.window().getStart()));
                        bean.setEdt(AtguiguUtil.toDatTime(ctx.window().getEnd()));
                        
                        bean.setTs(System.currentTimeMillis());
                        
                        out.collect(bean);
                        
                    }
                }
            );
        
    }
    
    private SingleOutputStreamOperator<TradeSkuOrderBean> distinctByOrderDetailId(
        SingleOutputStreamOperator<TradeSkuOrderBean> beanStream) {
        return beanStream
            .keyBy(TradeSkuOrderBean::getOrderDetailId)
            .process(new KeyedProcessFunction<String, TradeSkuOrderBean, TradeSkuOrderBean>() {
                
                private ValueState<TradeSkuOrderBean> lastBeanState;
                
                @Override
                public void open(Configuration parameters) throws Exception {
                    lastBeanState = getRuntimeContext().getState(new ValueStateDescriptor<TradeSkuOrderBean>("lastBeanState", TradeSkuOrderBean.class));
                }
                
                @Override
                public void processElement(TradeSkuOrderBean bean,
                                           Context ctx,
                                           Collector<TradeSkuOrderBean> out) throws Exception {
                    TradeSkuOrderBean lastBean = lastBeanState.value();
                    
                    if (lastBean == null) {  // 同一个详情id 的第一条数据过来, 直接输出
                        out.collect(bean);
                        // 存入到状态中
                        lastBeanState.update(bean);
                    } else {
                        
                        TradeSkuOrderBean newBean = new TradeSkuOrderBean();
                        BeanUtils.copyProperties(newBean, bean);  // 把 bean 中所有属性的值复制到 newBean 中
                        lastBeanState.update(newBean);  // 更新修改前的数据
                        
                        // 第二条数据进来. bean如果直接修改,
                        bean.setOriginalAmount(bean.getOriginalAmount().subtract(lastBean.getOriginalAmount()));
                        bean.setActivityAmount(bean.getActivityAmount().subtract(lastBean.getActivityAmount()));
                        bean.setCouponAmount(bean.getCouponAmount().subtract(lastBean.getCouponAmount()));
                        bean.setOrderAmount(bean.getOrderAmount().subtract(lastBean.getOrderAmount()));
                        out.collect(bean);
                    }
                }
            });
    }
    
    private SingleOutputStreamOperator<TradeSkuOrderBean> parseToPojo(DataStreamSource<String> stream) {
        return stream.map(new MapFunction<String, TradeSkuOrderBean>() {
            @Override
            public TradeSkuOrderBean map(String value) throws Exception {
                JSONObject obj = JSON.parseObject(value);
                
                BigDecimal splitActivityAmount = obj.getBigDecimal("split_activity_amount");
                BigDecimal splitCouponAmount = obj.getBigDecimal("split_coupon_amount");
                
                return TradeSkuOrderBean.builder()
                    .skuId(obj.getString("sku_id"))
                    .originalAmount(obj.getBigDecimal("split_original_amount"))
                    .activityAmount(splitActivityAmount == null ? new BigDecimal(0) : splitActivityAmount)
                    .couponAmount(splitCouponAmount == null ? new BigDecimal(0) : splitCouponAmount)
                    .orderAmount(obj.getBigDecimal("split_total_amount"))
                    .ts(obj.getLong("ts") * 1000)
                    .orderDetailId(obj.getString("id"))
                    .build();
            }
        });
    }
}
/*
异步超时异常....
其他的原因, 导致数据没有得到, 最后在 60s 内没有处理完这条数据, 导致异步超时
1. 检测用到的集群释放都已经起来
    hdfs hbase redis kafka mysql

2. 最后在测试异步之前,前面的没有使用异步的时候的代码测试通过

3. 查看 6 张维度表是否都在, 是否都有数据. 如果不是很确定, 最好使用 maxwell-boostrap 把 6 张表的维度全部重新导入一下

4. 异步超时时间加大 120

5. 找我

kafka集群修复:
    恢复一下 先停止 kakfa
    1. 删除 kafka 的数据
        xcall rm -rf /opt/module/kafka-3..0/logs/
    2. 去 zkClient 上删除 kakfa 节点
        deleteall /kafka
        
hbase集群修复:
    恢复
    
    1. 去 hdsf 上删除目录: /hbase
    
    2. 去 zkClient 上删除 /hbase 节点
      deleteall /hbase





-------
去 join 的时候都是使用同步的方式读, 效率比较低

同步


异步

使用异步 io 的先决条件:
 得有异步客户端
 
 redis 和 phoenix 没有异步客户端!!!
 
 多线程+多客户度
    每查询一个维度时候, 创建一个子线程(线程池),在这个子线程内创建一个客户端(连接池)


 */